package com.qch.kafka.demo.interceptor;

import com.qch.kafka.demo.entity.User;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Headers;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.logging.Logger;

/**
 * @Author qch
 * @Date 2021/5/31
 */
public class InterceptorTwo implements ProducerInterceptor<String,User> {
   // private static final Logger LOGGER = (Logger) LoggerFactory.getLogger(InterceptorTwo.class);
    public ProducerRecord<String, User> onSend(ProducerRecord<String, User> producerRecord) {
        System.out.println("拦截器2-》go");
        //消息发送的时候，经过拦截器，调用该方法
        //对原消息没有任何改动
        String topic = producerRecord.topic();
        Integer partition = producerRecord.partition();
        String key = producerRecord.key();
        User value = producerRecord.value();
        Long timestamp = producerRecord.timestamp();
        Headers headers = producerRecord.headers();
        ProducerRecord<String, User> newRecord = new ProducerRecord<String, User>(topic, partition, timestamp, key, value, headers);
        return newRecord;
    }

    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        System.out.println("拦截器2-》back");
        //消息确认或异常的时候，调用该方法，该方法中不应该实现较重的任务，会影响kafka生产者的性能
    }

    public void close() {

    }

    public void configure(Map<String, ?> configs) {
        Object classContent = configs.get("classContent");
        System.out.println(classContent);
    }
}
